package com.atguigu.etlinterceptor;

import com.alibaba.fastjson.JSON;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.util.Iterator;
import java.util.List;

public class ETLInterceptor implements Interceptor {
    /*** 初始化:拦截器有可能和mysql打交道,那么得创建连接 */
    @Override
    public void initialize() {
    }
    /**  * 拦截不是json格式的event
     * @param event
     * @return */
    @Override
    public Event intercept(Event event) {
        //取出数据
        byte[] body = event.getBody();
        String s = new String(body);
        //判断数据是否为json字符串,不是会报错,报错就过滤
        try {
            JSON.parseObject(s);
            return event;
        } catch (Exception e) {
            return null;
        }
    }
    @Override
    public List<Event> intercept(List<Event> events) {
        //采用迭代器删除集合中不是json字符串格式的数据
        Iterator<Event> iterator = events.iterator();
        while (iterator.hasNext()){
            Event next = iterator.next();
            if(intercept(next)==null){
                iterator.remove();
            }
        }
        return events;
    }
    @Override
    public void close() {
        //连接使用完成后在这里关闭
    }
    /*拦截器内部提供的builder内部类,用来构建自定义拦截对象*/
    public static class Builder implements Interceptor.Builder{
        @Override
        public Interceptor build() {
            return new ETLInterceptor();
        }
        @Override//获取拦截器配置
        public void configure(Context context) {}
    }
}
